#[cfg(target_family = "unix")]
#[cfg(target_family = "unix")]
use std::fs::OpenOptions;
use std::io;
use std::io::{Read, Write};
use std::net::TcpStream;
use std::path::{Path, PathBuf};

use anyhow::Result;
use indicatif::{MultiProgress, ProgressBar, ProgressStyle};
use ssh2::Session;
use tokio::sync::mpsc;
use tokio::task;
use tokio::task::JoinSet;

use crate::{error, log};
use crate::service::ServerItem;
use crate::utils::MAX_CONCURRENT;

pub struct SshUtil {
    pub session: Session,
}

impl SshUtil {
    pub fn login_server(server: &ServerItem) -> Result<Self> {
        let mut session = Session::new()?;
        let tcp_stream = TcpStream::connect(format!("{}:{}", server.ip, server.port))?;
        session.set_tcp_stream(tcp_stream);
        session.set_compress(true);
        session.set_timeout(30000);
        session.handshake()?;

        if server.auth_type == 0 {
            session.userauth_password(&server.user_name, &server.password)?;
        } else {
            session.userauth_pubkey_file(&server.user_name, None, Path::new(&server.secret_key), None)?;
        }
        Ok(Self { session })
    }

    pub fn exec(&mut self, cmd: String) -> Result<()> {
        log(format!("执行命令：{}", cmd));
        let mut channel = self.session.channel_session()?;
        channel.exec(&cmd)?;

        let mut buf = [0; 1024];
        loop {
            let len = channel.read(&mut buf)?;
            if len == 0 { break; }
            io::stdout().write_all(&buf[..len])?;
        }

        channel.send_eof()?;
        channel.wait_eof()?;
        channel.wait_close()?;

        channel.exit_status()?;
        Ok(())
    }

    pub async fn upload_file(&mut self, file_path: &Path, remote_path: &Path, remote_file: &Path) -> Result<()> {
        self.check_remote_dir(remote_path)?;
        let pb = ProgressBar::new(0);
        Self::copy_file(self.session.clone(), file_path.to_path_buf(), remote_file.to_path_buf(), pb, true).await
    }

    pub fn check_remote_dir(&mut self, path: &Path) -> Result<()> {
        self.exec(format!("mkdir -p {}", path.to_string_lossy()))
    }

    pub fn get_remote_dir(&mut self, remote: &Path) -> Result<Vec<PathBuf>> {
        let sftp = self.session.sftp()?;
        let mut remote_files = Vec::new();
        if let Ok(fs) = sftp.stat(remote) {
            if fs.is_file() {
                remote_files.push(remote.to_path_buf());
            } else {
                for (path, stat) in sftp.readdir(remote)? {
                    if stat.is_file() {
                        remote_files.push(path);
                    } else {
                        remote_files.extend(self.get_remote_dir(&path)?);
                    }
                }
            }
        }
        Ok(remote_files)
    }

    pub async fn multiple_scp(&mut self, files: Vec<(PathBuf, PathBuf)>, is_upload: bool) -> Result<()> {
        let m = MultiProgress::new();

        let len = files.len();
        let pb_main = m.add(ProgressBar::new(len as u64));
        pb_main.set_style(ProgressStyle::default_bar()
            .template("{spinner:.green}[{elapsed}][{bar:50.cyan/blue}] {pos:>7}/{len:7} {msg}")?
            .progress_chars("#>-"));
        pb_main.set_message(if is_upload { "上传中..." } else { "下载中..." });

        let mut tasks = JoinSet::new();
        let mut last_item = false;

        for (index, (local, remote)) in files.iter().enumerate() {
            if index == files.len() - 1 {
                last_item = true;
            }
            let pb_task = m.insert_before(&pb_main, ProgressBar::new(len as u64));
            tasks.spawn(Self::copy_file(self.session.clone(), local.clone(), remote.clone(), pb_task, is_upload));
            while tasks.len() >= MAX_CONCURRENT || last_item {
                match tasks.join_next().await {
                    Some(res) => {
                        if let Err(err) = res {
                            let msg = if is_upload {
                                format!("上传失败：{}", err)
                            } else {
                                format!("下载失败：{}", err)
                            };
                            error(msg);
                        }
                    }
                    None => { break; }
                };
                pb_main.inc(1);
            }
        }
        pb_main.finish_with_message(if is_upload { "上传完成！" } else { "下载完成！" });
        Ok(())
    }

    async fn copy_file(session: Session, local: PathBuf, remote: PathBuf, pb: ProgressBar, is_upload: bool) -> Result<()> {
        let capacity = 1024 * 1024;
        if is_upload {
            let fs_read = OpenOptions::new().read(true).open(&local)?;
            let len = fs_read.metadata()?.len();
            let file_name = local.file_name().expect("获取文件名错误！").to_string_lossy().to_string();
            pb.set_length(len);
            pb.set_style(ProgressStyle::default_bar()
                .template(&format!("{{spinner:.green}}[{{elapsed}}][{{bar:50.cyan/blue}}] {{bytes}}/{{total_bytes}}({{bytes_per_sec}},{{eta}}) {0}", file_name))?
                .progress_chars("#>-"));

            match session.scp_send(&remote, 0o644, len, None) {
                Ok(mut writer) => {
                    let input_file = io::BufReader::with_capacity(capacity, fs_read);
                    std::io::copy(&mut pb.wrap_read(input_file), &mut writer)?;
                    writer.send_eof()?;
                    writer.wait_eof()?;
                    writer.close()?;
                    writer.wait_close()?;
                }
                Err(err) => { println!("==== {}", err) }
            }
        } else {
            let (mut fs_read, stat) = session.scp_recv(&remote)?;
            let len = stat.size();
            let file_name = remote.to_string_lossy().to_string();
            pb.set_length(len);
            pb.set_style(ProgressStyle::default_bar()
                .template(&format!("{{spinner:.green}}[{{elapsed}}][{{bar:50.cyan/blue}}] {{bytes}}/{{total_bytes}}({{bytes_per_sec}},{{eta}}) {0}", file_name))?
                .progress_chars("#>-"));
            let mut fs_writer = OpenOptions::new().write(true).create_new(true).truncate(true).open(&local)?;
            let input_file = io::BufReader::with_capacity(capacity, &mut fs_read);
            std::io::copy(&mut pb.wrap_read(input_file), &mut fs_writer)?;
            fs_read.send_eof()?;
            fs_read.wait_eof()?;
            fs_read.close()?;
            fs_read.wait_close()?;
        };
        pb.finish_with_message("done!");
        Ok(())
    }

    pub async fn auto_exec(&mut self, cmds: Vec<String>, dir: Option<String>) -> Result<()> {
        let mut channel = self.session.channel_session()?;
        channel.request_pty("xterm", None, None)?;
        channel.handle_extended_data(ssh2::ExtendedData::Merge)?;
        channel.shell()?;
        self.session.set_blocking(false);

        let (tx, mut rx) = mpsc::channel(102);

        let producer = task::spawn(async move {
            if let Some(dir) = dir {
                tx.send(format!("cd {} \n", dir)).await.unwrap();
            }
            for cmd in cmds {
                tx.send(format!("{} \n", cmd)).await.unwrap();
            }
            tx.send("exit \n".to_string()).await.unwrap();
        });
        producer.await?;

        let mut buffer = [0u8; 4096];
        let stdout = io::stdout();
        let mut handle = stdout.lock();
        loop {
            match channel.read(&mut buffer) {
                Ok(size) => {
                    if String::from_utf8_lossy(&buffer).contains("logout") {
                        break;
                    }
                    handle.write_all(&buffer[0..size])?;
                    handle.flush()?;
                }
                Err(e) => {
                    if e.kind() != io::ErrorKind::WouldBlock {
                        handle.write_all(e.to_string().as_bytes())?;
                        handle.flush()?;
                    }
                }
            }
            if let Some(line) = rx.recv().await {
                channel.write_all(line.as_bytes())?;
                channel.flush()?;
            }
        }
        handle.write_all(b"\n")?;
        handle.flush()?;
        Ok(())
    }
}